package cn.com.sparkle.paxos.paxosinstance;

import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

import org.apache.log4j.Logger;

import com.google.protobuf.GeneratedMessage;

//import cn.com.sparkle.paxos.model.Id;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Value;
import cn.com.sparkle.paxos.net.MajorSender;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.net.callback.PrepareCallBack;
import cn.com.sparkle.paxos.net.callback.VoteCallBack;

public abstract class PaxosInstance {
	private final static Logger logger = Logger.getLogger(PaxosInstance.class);
	/**
	 * if conflictId is -1,it may indicates that response from node is not
	 * enough half because of timeout.
	 */
	private volatile Set<NetNode> activeNode;
	private int quorum;
	protected long instanceId;
	// protected Id id;

	private Value lastValue;
	private volatile Boolean result;
	private FutureTask<Boolean> future;
	private String selfAddress;

	public PaxosInstance(Set<NetNode> activeNode, int quorum, long instanceId,
			String selfAddress) {
		super();
		this.activeNode = activeNode;
		this.quorum = quorum;
		this.instanceId = instanceId;
		// this.id = id;
		this.selfAddress = selfAddress;
	}

	public Future<Boolean> activate() {
		// CountDownLatch countDownLatch = new CountDownLatch(quorum);
		future = new FutureTask<Boolean>(new Callable<Boolean>() {
			@Override
			public Boolean call() throws Exception {
				return result;
			}
		});
		PrepareCallBack prepareCallBack = new PrepareCallBack(quorum,
				activeNode.size(), this);
		MajorSender.sendToMajor(activeNode, getPrepareRequest(),
				prepareCallBack);

		return future;
	}

	public void resetActiveNode(Set<NetNode> activeNode) {
		this.activeNode = activeNode;
	}

	public void fail(long refuseId) {
		result = false;
		instanceFail(refuseId);
		future.run();
	}

	public final void prepareSuccess(Value returnLastValue) {
		logger.debug("prepareSuccessEvent");
		lastValue = returnLastValue != null ? returnLastValue
				: getWantAssginValue();
		VoteCallBack voteCallBack = new VoteCallBack(quorum, activeNode.size(),
				this);
		MajorSender.sendToMajor(activeNode, getVoteRequest(lastValue),
				voteCallBack);
	}

	public final long getInstanceId() {
		return instanceId;
	}

	public final void voteSuccess() {
		logger.debug("voteSuccess instanceId:" + instanceId);
		voteSuccess(lastValue);
		MajorSender.sendToMajorExceptSelf(activeNode,
				getSuccessMessage(lastValue), null, selfAddress);
		result = true;
		future.run();
	}

	public int getQuorum() {
		return quorum;
	}

	public abstract GeneratedMessage getSuccessMessage(Value lastValue);

	public abstract GeneratedMessage getPrepareRequest();

	public abstract GeneratedMessage getVoteRequest(Value lastValue);

	public abstract Value getWantAssginValue();

	public abstract void voteSuccess(Value value);

	/**
	 * 
	 * @param refuseId
	 *            -1 indicates the number of active senator node is not enough
	 */
	public abstract void instanceFail(long refuseId);

}
